package com.wugui.datax.job.service.impl;

import com.github.CCweixiao.HBaseTemplate;
import com.wugui.datax.job.entity.StudentScore;
import com.wugui.datax.job.mapper.base.HbaseMapper;
import com.wugui.datax.job.mapper.base.HiveMapper;
import com.wugui.datax.job.mapper.base.RdbmsMapper;
import com.wugui.datax.job.service.StudentService;
import com.wugui.datax.job.tool.query.RdbmsTemplate;
import com.wugui.datax.job.util.JdbcConstants;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;

@Service
public class StudentServiceImpl implements StudentService {

    @Autowired
    private RdbmsMapper rdbmsMapper;

    @Autowired
    private HiveMapper hiveMapper;

    @Autowired
    private HbaseMapper hbaseMapper;

    @Override
    public void extract() throws Exception {
        RdbmsTemplate rdbmsTemplate = rdbmsMapper.rdbmsTemplate(JdbcConstants.DATASOURCE_DEMO);
        // 查询昨天到当前的学生
        ResultSet rs = rdbmsTemplate.executeQuery("select id,no,name,IF(sex=1,'男','女') sex,height,weight from t_student");
        //此处没考虑效率,java代码通常只用来处理比较复杂的逻辑，数据源非数据库，datax处理不了的场景，只是演示功能。
        PreparedStatement stmt =  hiveMapper.hiveTemplate().preparedStatement("insert into t_student values(?,?,?,?,?,?,?)");
        //清除表

        hiveMapper.hiveTemplate().executeUpdate("truncate table t_student");
        while (rs.next()){
            stmt.setInt(1,rs.getInt(1));
            stmt.setString(2,rs.getString(2));
            stmt.setString(3,rs.getString(3));
            stmt.setString(4,rs.getString(4));
            stmt.setString(5,rs.getString(5));
            stmt.setString(6,rs.getString(6));
            stmt.setLong(7,System.currentTimeMillis());
            stmt.execute();
          }
        rs.close();
        stmt.close();
    }

    @Override
    public void trans() throws Exception {
        ResultSet rs = hiveMapper.hiveTemplate().executeQuery("select a.stu_no,a.stu_name,sum(b.score) score  from  t_student a ,t_course_score b where a.stu_no=b.stu_no group by a.stu_no,a.stu_name");
        List<StudentScore> datalist = new ArrayList<>();
        while(rs.next()){
            StudentScore studentScore = hbaseMapper.hBaseTemplate().getByRowKey(rs.getString(1),StudentScore.class);
            if(studentScore==null){
                StudentScore data = new StudentScore();
                data.setNo(rs.getString(1));
                data.setName(rs.getString(2));
                data.setScore(rs.getDouble(3));
                datalist.add(data);
            }

        }
        if(datalist.size()>0){
            hbaseMapper.hBaseTemplate().saveBatch(datalist);
        }
        rs.close();

    }
}
